package com.ksyun.campus.metaserver.services;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.metaserver.config.MinfsProperties;
import com.ksyun.campus.metaserver.domain.MetaServerInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;

import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * 增强的 Leader 选举服务
 * 支持：
 * 1. ZooKeeper 临时顺序节点选举
 * 2. Leader 故障检测与自动故障转移
 * 3. 元数据同步协调
 * 4. 集群状态监控
 */
@Slf4j
@Service
public class LeaderElectionService {
    
    @Autowired
    private CuratorFramework curator;
    
    @Autowired
    private MinfsProperties properties;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    @Value("${server.port:8300}")
    private int serverPort;
    
    private LeaderLatch leaderLatch;
    private PersistentEphemeralNode instanceNode;
    private volatile boolean isLeader = false;
    private volatile boolean isInitialized = false;
    private final AtomicBoolean isShuttingDown = new AtomicBoolean(false);
    
    // 选举相关路径
    private String leaderElectionPath;
    private String instancesPath;
    private String currentInstancePath;
    
    // 实例信息
    private String instanceId;
    private String hostAddress;
    private MetaServerInfo currentInstance;
    
    // 监听器
    private final CopyOnWriteArrayList<Consumer<Boolean>> leadershipListeners = new CopyOnWriteArrayList<>();
    private final CopyOnWriteArrayList<Consumer<MetaServerInfo>> leaderChangeListeners = new CopyOnWriteArrayList<>();
    
    // 集群状态
    private final ConcurrentHashMap<String, MetaServerInfo> clusterMembers = new ConcurrentHashMap<>();
    private final CountDownLatch initializationLatch = new CountDownLatch(1);
    
    @PostConstruct
    public void initialize() throws Exception {
        log.info("初始化 Leader 选举服务");
        
        // 初始化基本信息
        instanceId = properties.getCluster().getInstanceId();
        hostAddress = InetAddress.getLocalHost().getHostAddress();
        
        // 初始化路径
        String rootPath = properties.getZookeeper().getRootPath();
        leaderElectionPath = rootPath + "/meta/leader-election";
        instancesPath = rootPath + "/meta/instances";
        currentInstancePath = instancesPath + "/" + instanceId;
        
        // 创建基础路径
        ensureBasePaths();
        
        // 创建当前实例信息
        createCurrentInstance();
        
        // 注册实例节点
        registerInstance();
        
        // 启动 Leader 选举
        startLeaderElection();
        
        // 添加连接状态监听器
        addConnectionStateListener();
        
        isInitialized = true;
        initializationLatch.countDown();
        
        log.info("Leader 选举服务初始化完成，实例ID: {}", instanceId);
    }
    
    @PreDestroy
    public void shutdown() throws Exception {
        if (isShuttingDown.getAndSet(true)) {
            return;
        }
        
        log.info("关闭 Leader 选举服务");
        
        try {
            if (leaderLatch != null) {
                leaderLatch.close();
            }
            
            if (instanceNode != null) {
                instanceNode.close();
            }
            
            clusterMembers.clear();
            leadershipListeners.clear();
            leaderChangeListeners.clear();
            
        } catch (Exception e) {
            log.error("关闭 Leader 选举服务时出错", e);
        }
        
        log.info("Leader 选举服务已关闭");
    }
    
    private void ensureBasePaths() throws Exception {
        String[] paths = {
                properties.getZookeeper().getRootPath(),
                properties.getZookeeper().getRootPath() + "/meta",
                leaderElectionPath,
                instancesPath
        };
        
        for (String path : paths) {
            if (curator.checkExists().forPath(path) == null) {
                curator.create()
                        .creatingParentsIfNeeded()
                        .forPath(path);
                log.debug("创建 ZK 路径: {}", path);
            }
        }
    }
    
    private void createCurrentInstance() {
        currentInstance = MetaServerInfo.builder()
                .instanceId(instanceId)
                .host(hostAddress)
                .port(serverPort)
                .isLeader(false)
                .startTime(System.currentTimeMillis())
                .lastHeartbeat(LocalDateTime.now())
                .build();
    }
    
    private void registerInstance() throws Exception {
        // 注册实例信息（持久化临时节点）
        String instanceData = objectMapper.writeValueAsString(currentInstance);
        
        instanceNode = new PersistentEphemeralNode(
                curator,
                PersistentEphemeralNode.Mode.EPHEMERAL,
                currentInstancePath,
                instanceData.getBytes(StandardCharsets.UTF_8)
        );
        
        instanceNode.start();
        instanceNode.waitForInitialCreate(30, TimeUnit.SECONDS);
        
        log.info("注册实例节点: {} -> {}", currentInstancePath, instanceData);
    }
    
    private void startLeaderElection() throws Exception {
        leaderLatch = new LeaderLatch(curator, leaderElectionPath, instanceId);
        
        leaderLatch.addListener(new LeaderLatchListener() {
            @Override
            public void isLeader() {
                try {
                    handleBecomeLeader();
                } catch (Exception e) {
                    log.error("处理成为 Leader 事件时出错", e);
                }
            }
            
            @Override
            public void notLeader() {
                try {
                    handleLoseLeadership();
                } catch (Exception e) {
                    log.error("处理失去 Leader 身份事件时出错", e);
                }
            }
        });
        
        leaderLatch.start();
        log.info("启动 Leader 选举，实例ID: {}, 选举路径: {}", instanceId, leaderElectionPath);
        
        // 等待一段时间让选举稳定
        Thread.sleep(2000);
        log.info("Leader 选举启动完成，当前状态: {}", leaderLatch.hasLeadership() ? "Leader" : "Follower");
    }
    
    private void handleBecomeLeader() throws Exception {
        log.info("====== 当前实例成为 Leader ======");
        isLeader = true;
        
        // 更新实例信息
        currentInstance.setLeader(true);
        currentInstance.setLeaderElectedTime(LocalDateTime.now());
        updateInstanceData();
        
        // 通知监听器
        notifyLeadershipListeners(true);
        notifyLeaderChangeListeners(currentInstance);
        
        log.info("Leader 初始化完成，开始提供服务");
    }
    
    private void handleLoseLeadership() throws Exception {
        log.info("====== 当前实例失去 Leader 身份 ======");
        isLeader = false;
        
        // 更新实例信息
        currentInstance.setLeader(false);
        currentInstance.setLeaderElectedTime(null);
        updateInstanceData();
        
        // 通知监听器
        notifyLeadershipListeners(false);
        
        log.info("已切换为 Follower 模式");
    }
    
    private void updateInstanceData() throws Exception {
        if (instanceNode != null) {
            String instanceData = objectMapper.writeValueAsString(currentInstance);
            instanceNode.setData(instanceData.getBytes(StandardCharsets.UTF_8));
        }
    }
    
    private void addConnectionStateListener() {
        curator.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                log.info("ZooKeeper 连接状态变化: {}", newState);
                
                switch (newState) {
                    case LOST:
                        log.error("ZooKeeper 连接丢失，停止 Leader 活动");
                        if (isLeader) {
                            // 临时失去领导权，但不改变 isLeader 状态，等待重连
                            notifyLeadershipListeners(false);
                        }
                        break;
                        
                    case RECONNECTED:
                        log.info("ZooKeeper 重新连接成功");
                        // 重连后状态会通过 LeaderLatch 自动恢复
                        break;
                        
                    case SUSPENDED:
                        log.warn("ZooKeeper 连接挂起");
                        break;
                        
                    default:
                        log.debug("ZooKeeper 连接状态: {}", newState);
                        break;
                }
            }
        });
    }
    
    /**
     * 等待初始化完成
     */
    public boolean waitForInitialization(long timeout, TimeUnit unit) {
        try {
            return initializationLatch.await(timeout, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
    
    /**
     * 检查当前实例是否为 Leader
     */
    public boolean isLeader() {
        return isLeader && isInitialized && !isShuttingDown.get();
    }
    
    /**
     * 获取当前 Leader 信息
     */
    public MetaServerInfo getCurrentLeader() throws Exception {
        if (!isInitialized) {
            return null;
        }
        
        try {
            String leaderId = leaderLatch.getLeader().getId();
            if (leaderId != null) {
                String leaderInstancePath = instancesPath + "/" + leaderId;
                if (curator.checkExists().forPath(leaderInstancePath) != null) {
                    byte[] data = curator.getData().forPath(leaderInstancePath);
                    return objectMapper.readValue(new String(data, StandardCharsets.UTF_8), MetaServerInfo.class);
                }
            }
        } catch (Exception e) {
            log.warn("获取 Leader 信息失败", e);
        }
        
        return null;
    }
    
    /**
     * 获取当前实例信息
     */
    public MetaServerInfo getCurrentInstance() {
        return currentInstance;
    }
    
    /**
     * 强制放弃 Leader 身份（用于测试）
     */
    public void releaseLeadership() throws Exception {
        if (leaderLatch != null && leaderLatch.hasLeadership()) {
            log.info("主动放弃 Leader 身份");
            leaderLatch.close();
            // 重新创建 LeaderLatch 以重新参与选举
            startLeaderElection();
        }
    }
    
    /**
     * 添加 Leader 身份变化监听器
     */
    public void addLeadershipListener(Consumer<Boolean> listener) {
        leadershipListeners.add(listener);
    }
    
    /**
     * 添加 Leader 变更监听器
     */
    public void addLeaderChangeListener(Consumer<MetaServerInfo> listener) {
        leaderChangeListeners.add(listener);
    }
    
    /**
     * 移除监听器
     */
    public void removeLeadershipListener(Consumer<Boolean> listener) {
        leadershipListeners.remove(listener);
    }
    
    public void removeLeaderChangeListener(Consumer<MetaServerInfo> listener) {
        leaderChangeListeners.remove(listener);
    }
    
    private void notifyLeadershipListeners(boolean isLeader) {
        for (Consumer<Boolean> listener : leadershipListeners) {
            try {
                listener.accept(isLeader);
            } catch (Exception e) {
                log.error("通知 Leader 身份变化监听器失败", e);
            }
        }
    }
    
    private void notifyLeaderChangeListeners(MetaServerInfo leader) {
        for (Consumer<MetaServerInfo> listener : leaderChangeListeners) {
            try {
                listener.accept(leader);
            } catch (Exception e) {
                log.error("通知 Leader 变更监听器失败", e);
            }
        }
    }
    
    /**
     * 获取集群成员信息
     */
    public ConcurrentHashMap<String, MetaServerInfo> getClusterMembers() {
        return new ConcurrentHashMap<>(clusterMembers);
    }
    
    /**
     * 健康检查
     */
    public boolean isHealthy() {
        return isInitialized && 
               !isShuttingDown.get() && 
               curator.getState().toString().equals("STARTED") &&
               leaderLatch != null;
    }
}
